package com.mai.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mai.realtime.app.BaseAppV1;
import com.mai.realtime.bean.TrafficHomeDetailPageViewBean;
import com.mai.realtime.common.Constant;
import com.mai.realtime.util.AtguiguUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.time.Duration;

/**
 * com.mai.realtime.app.dws.DwsTrafficPageViewWindow
 * 页面浏览数
 * 写入clickhouse dws_traffic_page_view_window
 */
public class DwsTrafficPageViewWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsTrafficPageViewWindow().init(
                4003,
                1,
                "DwsTrafficPageViewWindow",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1、找到详情也和首页，封装bean
        //统计当日的首页和商品详情页独立访客
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> beanStream = findUv(stream);
        //  2、 开窗(全窗口，滚动5s一个窗口)聚合（reduce）
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> resultStream = windowAndAgg(beanStream);
        //resultStream.print();
        // 3:写出到clickhouse
        writeToClickhouse(resultStream);

    }

    private void writeToClickhouse(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> resultStream) {
        resultStream.addSink(new RichSinkFunction<TrafficHomeDetailPageViewBean>() {

            private PreparedStatement insertStmt;
            private Connection conn;

            @Override
            public void open(Configuration parameters) throws Exception {
                //conn = DriverManager.getConnection("jdbc:clickhouse://hadoop162:3579/default");
                conn = DriverManager.getConnection("jdbc:clickhouse://hadoop162:8123/default");
                //(stt,edt,curDate,home_uv_ct,good_detail_uv_ct,ts)
                insertStmt = conn.prepareStatement("INSERT INTO dws_traffic_page_view_window  VALUES (?,?,?,?,?,?)");
                //dws_traffic_page_view_window
                //dws_traffic_page_view_window
            }

            @Override
            public void close() throws Exception {
                insertStmt.close();
                conn.close();
            }

            @Override
            public void invoke(TrafficHomeDetailPageViewBean value, Context ctx) throws Exception {
                insertStmt.setString(1, value.getStt());
                insertStmt.setString(2, value.getEdt());
                insertStmt.setString(3, value.getCurDate());
                insertStmt.setString(4, String.valueOf(value.getHomeUvCt()));
                insertStmt.setString(5, String.valueOf(value.getGoodDetailUvCt()));
                insertStmt.setString(6, String.valueOf(value.getTs()));

/*                insertStmt.setLong(4, value.getHomeUvCt());
                insertStmt.setLong(5, value.getGoodDetailUvCt());
                insertStmt.setLong(6, value.getTs());*/
            }
        });
    }

    private SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> windowAndAgg(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<TrafficHomeDetailPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(((bean, ts) -> bean.getTs()))
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))//测试用5秒 实际看需求
                .reduce(new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                            @Override
                            public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean bean1,
                                                                        TrafficHomeDetailPageViewBean bean2) throws Exception {
                                bean1.setHomeUvCt(bean1.getHomeUvCt() + bean2.getHomeUvCt());
                                bean1.setGoodDetailUvCt(bean1.getGoodDetailUvCt() + bean2.getGoodDetailUvCt());

                                return bean1;
                            }
                        },
                        new ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
                            @Override
                            public void process(Context ctx,
                                                Iterable<TrafficHomeDetailPageViewBean> elements,
                                                Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                                TrafficHomeDetailPageViewBean bean = elements.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis()));

                                out.collect(bean);
                            }
                        }
                );

    }


    private SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> findUv(DataStreamSource<String> stream) {
        return stream
                .map(json-> JSON.parseObject(json))
                .filter(obj->{
                    String pageId = obj.getJSONObject("page").getString("page_id");
                    return  "home".equals(pageId)||"course_detail".equals(pageId);
                })
                // 统计home的独立访客数：把统一用户的记录分为一组
                .keyBy(obj->obj.getJSONObject("common").getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {

                    //TODO 状态：日期
                    private ValueState<String> goodDetailState;
                    private ValueState<String> homeState;

                    @Override
                    public void open(Configuration parameters) throws Exception {


                        //TODO 状态存储的是日期
                        // 如果当天的日期和状态的存储的日期不一样就是新的一天
                        homeState = getRuntimeContext().getState(new ValueStateDescriptor<String>("homeState", String.class));

                        goodDetailState = getRuntimeContext().getState(new ValueStateDescriptor<String>("goodDetailState", String.class));

                    }

                    @Override
                    public void processElement(JSONObject obj,
                                               Context ctx,
                                               Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                        String pageId = obj.getJSONObject("page").getString("page_id");

                        Long ts = obj.getLong("ts");

                        //TODO  统计当日的首页和商品详情页独立访客数

                        Long homeUvCt = 0L;
                        Long goodDetailUvCt = 0L;

                        String today = AtguiguUtil.toDate(ts);

                        // 是home, 并且今天和状态中的年月日不等, 就是今天的第一个home
                        if ("home".equals(pageId)&&!today.equals(homeState.value())) {
                            homeUvCt = 1L;
                            //TODO 记得更新状态
                            homeState.update(today);
                        }else if ("good_detail".equals(pageId) && !today.equals(goodDetailState.value())){

                            goodDetailUvCt = 1L;
                            goodDetailState.update(today);
                        }

                        if (homeUvCt+goodDetailUvCt>=1){
                            out
                                    .collect(
                                            new TrafficHomeDetailPageViewBean(
                                                    "", "",
                                                    "",
                                                    homeUvCt, goodDetailUvCt,
                                                    ts
                                            )
                                    );
                        }


                    }
                });


    }
}
